package bsf.baseservice.businessmq.consumer;

import java.util.LinkedList;
import java.util.Map;

/**
 消息分区队列 多线程安全
 
*/
public class PartitionQueue<T>
{
	private java.util.HashMap<Integer, LinkedList<T>> partitionQuque = new java.util.HashMap<Integer, LinkedList<T>>();
	private Object ququelock = new Object();

	public final void enqueue(int patition, java.util.ArrayList<T> Messages)
	{
		synchronized (ququelock)
		{
			LinkedList<T> q = getQueque(patition);
			for (T m : Messages)
			{
				q.offer(m);
			}
		}

	}

	public final T dequeue(int patition)
	{
		synchronized (ququelock)
		{
			LinkedList<T> q = getQueque(patition);
			if (q.size() != 0)
			{
				return q.poll();
			}
			else
			{
				return null;
			}

		}
	}


	public final void clear()
	{
		synchronized (ququelock)
		{
			partitionQuque.clear();
		}
	}
	/** 
	 获取所有分区中最大的队列数量（不安全方式）
	 
	*/
	public final int maxCountOfPartition()
	{
		int count = 0;
		try
		{
			for (Map.Entry<Integer,LinkedList<T>> p : partitionQuque.entrySet())
			{
				count = Math.max(p.getValue().size(),count);
			}
		}
		catch (RuntimeException exp)
		{
		}

		return count;
	}

	private LinkedList<T> getQueque(int partition)
	{
		if (!partitionQuque.containsKey(partition))
		{
			LinkedList<T> q = new LinkedList<T>();
			partitionQuque.put(partition, q);
		}

		return partitionQuque.get(partition);
	}

}